package com.cathyjava.easyflows.workflow.parallel;

import com.cathyjava.easyflows.work.Work;
import com.cathyjava.easyflows.work.WorkContext;
import com.cathyjava.easyflows.work.WorkReport;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

/**
 * @author chenjing
 * @date 2021/5/12
 * @description 关注下这里的实现
 */
public class ParallelFlowExecutor {
    private ExecutorService executorService;

    public ParallelFlowExecutor(ExecutorService executorService) {
        this.executorService = executorService;
    }

    public List<WorkReport> executeInParallel(List<Work> workUnits, WorkContext workContext) {
        List<Callable<WorkReport>> tasks = new ArrayList<>();
        workUnits.forEach(x -> {
            tasks.add(() -> x.execute(workContext));
        });

        List<Future<WorkReport>> futures = null;
        try {
            futures = executorService.invokeAll(tasks);
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new RuntimeException("并行单元执行出错", e);
        }

        Map<String, Future<WorkReport>> map = new HashMap<>();
        for (int i = 0; i < workUnits.size(); i++) {
            map.put(workUnits.get(i).getName(), futures.get(i));
        }

        List<WorkReport> workReports = new ArrayList<>();
        for (Map.Entry<String, Future<WorkReport>> entry : map.entrySet()) {
            try {
                workReports.add(entry.getValue().get());
            } catch (InterruptedException e) {
                String message = String.format("The parallel flow was interrupted while waiting for the result of work unit '%s'", entry.getKey());
                throw new RuntimeException(message, e);
            } catch (ExecutionException e) {
                String message = String.format("Unable to execute flow unit %s", entry.getKey());
                throw new RuntimeException(message, e);
            }
        }

        return workReports;
    }
}
